package com.fink.demo.wordcount;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * 统计字符出现次数(基于流处理、这种写法推荐、无界流 )
 *   hello world
 *   hello flink
 *   hello java
 */
public class WordCountStreamUnboundedBatch {

    public static void main(String[] args) throws Exception {

        // 1.获取环境
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();

        // 2.读取文件
        DataStreamSource<String> file = environment.socketTextStream("practive",7777);


        // 3.处理数据
        SingleOutputStreamOperator<Tuple2<String, Integer>> flatMap = file.flatMap((String s, Collector<Tuple2<String, Integer>> collector) -> {
            String[] s1 = s.split(" ");
            for (String item : s1) {
                Tuple2<String, Integer> of = Tuple2.of(item, 1);
                collector.collect(of);
            }
        }).returns(Types.TUPLE(Types.STRING,Types.INT)).keyBy(data -> data.f0).sum(1);

        // 4. 打印
        flatMap.print();

        // 5.执行
        environment.execute();
    }
}
